﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;

namespace O1.Kernel.Storage.Transactions
{
    [ContractVerification(true)]
    internal class MultiVersionSequence
    {
        private long counter;

        private long watermark;

        private HashSet<long> retiredEarly;

        private object sync = new object();

        public MultiVersionSequence(long recoveryCheckpoint)
            : this()
        {
            Contract.Requires(recoveryCheckpoint > 0);

            this.counter = recoveryCheckpoint.AsPending();
            this.watermark = recoveryCheckpoint.AsCommitted();
        }

        public MultiVersionSequence()
        {
            this.retiredEarly = new HashSet<long>();
        }

        public long Watermark
        {
            get
            {
                return this.watermark;
            }
        }

        public long NextVersion()
        {
            Contract.Ensures(Contract.Result<long>() > 0);

            return Interlocked.Increment(ref this.counter);
        }

        public void RetireVersion(long versionId)
        {
            // A long running Tx can block the counter for some time, and transactions can be retired out of order.
            // An expectation must be set for outstanding transactions that are not retired.
            // Spec:
            //      1. If the target TX to retire is sequential to the last to be retired, all is well. Happy path.
            //      2. If the target TX is less than the last to be retired, there is a serious Tx sync problem. Throw?
            //      3. If the target TX is greater than the next to retire (retired +1)... it needs to be known as committed.
            versionId = versionId.AsCommitted();
            lock (this.sync)
            {
                versionId = versionId.AsPending();
                if (this.watermark + 1 == versionId)
                {
                    this.watermark = versionId;
                    return;
                }

                this.retiredEarly.Add(versionId);

                var next = this.watermark + 1;
                for (;;)
                {
                    if (!this.retiredEarly.Contains(next))
                    {
                        break;
                    }

                    this.retiredEarly.Remove(next);
                    this.watermark = next;
                    next = this.watermark + 1;
                }
            }
        }

        public bool IsVisible(long versionId)
        {
            lock (this.sync)
            {
                var max = this.watermark;
                if (versionId <= max)
                {
                    return true;
                }

                if (this.retiredEarly.Contains(versionId))
                {
                    return true;
                }
            }

            return false;
        }

        [ContractInvariantMethod]
        private void ObjectInvariant()
        {
            Contract.Invariant(this.counter > 0);
        }
    }
}